package com.lei.study.depth.lock;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * <p>
 * Zookeeper 实现分布式锁
 * </p>
 *
 * @author <a href="mailto:17674030991@163.com">伍磊</a>
 */
public class ZookeeperLock implements Lock {

    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperLock.class);

    private ZookeeperClient zookeeperClient;

    private String curNodeName;

    public ZookeeperLock() {
        zookeeperClient = ZookeeperClient.getInstance();
    }

    @Override
    public void lock() {
        try {
            doLock(false);
        } catch (InterruptedException e) {
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        doLock(true);
    }

    @Override
    public boolean tryLock() {
        try {
            // 1. 创建节点
            curNodeName = zookeeperClient.createNode();

            // 2. 获取最小节点
            String youngestNode = zookeeperClient.getYoungestNode();

            // 3. 比较节点
            return StringUtils.equals(curNodeName, youngestNode);
        } catch (Exception ignored) {
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return doLock(time, unit, false);
    }

    @Override
    public void unlock() {
        if (StringUtils.isNotBlank(curNodeName)) {
            try {
                zookeeperClient.delNode(curNodeName);
            } catch (Exception e) {
                LOG.error("删除节点异常: ", e);
            }
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    private void doLock(boolean interruptFlag) throws InterruptedException {
        doLock(null, null, interruptFlag);
    }

    private boolean doLock(Long time, TimeUnit unit, boolean interruptFlag) throws InterruptedException {
        // 定义超时时间
        AtomicLong atomicLong = new AtomicLong(Long.MAX_VALUE);
        if (null != time && null != unit) {
            // 转换为秒
            atomicLong.set(unit.toSeconds(time));
        }
        AtomicBoolean atLocked = new AtomicBoolean(tryLock());
        while (!atLocked.get()) {
            // 没抢到锁
            try {
                // 1. 获取前一个节点 -> 为空说明抢到了 -> 直接返回
                String preNode = zookeeperClient.getPreNode(curNodeName);
                if (StringUtils.isBlank(preNode)) {
                    break;
                }
                // 没抢到锁 查看是否被打断了
                if (interruptFlag && Thread.interrupted()) {
                    throw new InterruptedException();
                }

                // 2. 绑定监听事件
                CountDownLatch countDownLatch = new CountDownLatch(1);
                AtomicBoolean tryLocked = new AtomicBoolean(false);
                zookeeperClient.addWatch(preNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        if (event.getType() == Event.EventType.NodeDeleted) {
                            countDownLatch.countDown();
                            tryLocked.set(true);
                        }
                    }
                });
                // 等待2秒
                countDownLatch.await(2, TimeUnit.SECONDS);
                atomicLong.addAndGet(-2);
                if (atomicLong.get() <= 0) {
                    break;
                }
                if (tryLocked.get()) {
                    atLocked.set(true);
                }
            } catch (Exception e) {
                // 处理
                if (interruptFlag && e instanceof InterruptedException) {
                    throw new InterruptedException();
                }
            }
        }

        return atLocked.get();
    }
}
